package com.credithc.eagleeye.api.job;
import com.credithc.eagleeye.api.factory.ESFactory;
import com.credithc.eagleeye.api.utils.SendMailUtils;
import com.credithc.eagleeye.api.utils.WXWorkUtils;
import com.credithc.eagleeye.api.vo.AlertLogVo;
import com.credithc.eagleeye.api.vo.AlertRuleVo;
import com.credithc.eagleeye.vo.RpcTraceInfoVo;
import com.credithc.framework.utils.date.DateTools;
import com.credithc.framework.utils.json.FastJsonConvert;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.apache.commons.lang3.time.FastDateFormat;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.QueryStringQueryBuilder;
import org.elasticsearch.search.sort.SortBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
/**
 * 描 述: <br/>
 * 作 者: zhangjin<br/>
 * 创 建：2018年12月17日<br/>
 * 版 本：v1.0.0<br>
 * <p>
 * 历 史: (版本) 作者 时间 注释 <br/>
 */
@SuppressWarnings("Duplicates")
@Slf4j
@Component
public class AlertJobManager {
    @Autowired
    SendMailUtils sendMailUtils;
    @Autowired
    WXWorkUtils wxWorkUtils;
    private ConcurrentHashMap<String, ScheduledFuture<?>> taskMap = new ConcurrentHashMap<>();
    private ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(Runtime.getRuntime().availableProcessors() * 2,
            new BasicThreadFactory.Builder().namingPattern("alert-schedule-pool-%d").daemon(true).build());
    private static ReentrantLock lock = new ReentrantLock();
    private static final FastDateFormat ISO_DATETIME_TIME_ZONE_FORMAT_WITH_MILLIS = FastDateFormat.getInstance("yyyy-MM-dd'T'HH:mm:ss.SSSZZ");
    private final static String EAGLE_MONITOR_ES_INDEX = "eagle-monitor-*";
    private final static String EAGLE_MONITOR_ES_TYPE = "eagle";
    private final static String EAGLE_ALERT_LOG_ES_INDEX = ".eagle-alert-log";
    private final static String EAGLE_ALERT_LOG_ES_TYPE = "log";
    private ConcurrentHashMap<String, AlertStat> alertStatMap = new ConcurrentHashMap<>();

    public void addOrUpdate(AlertRuleVo alertRuleVo) {
        try {
            lock.lock();
            if (alertRuleVo != null && StringUtils.isNotBlank(alertRuleVo.getRuleId())) {
                String ruleId = alertRuleVo.getRuleId();
                cancel(ruleId);
                if (alertRuleVo.isEnable()) {
                    Runnable runnable = buildAlertJob(alertRuleVo);
                    ScheduledFuture<?> scheduledFuture = executorService.scheduleAtFixedRate(runnable, 30, alertRuleVo.getJobPeriodSeconds(), TimeUnit.SECONDS);
                    taskMap.putIfAbsent(ruleId, scheduledFuture);
                    log.info("添加了alertJob任务:{}", FastJsonConvert.convertObjectToJsonWithPrettyFormat(alertRuleVo));
                }
            }
        } catch (Exception ex) {
            log.error(ex.getMessage(), ex);
        } finally {
            lock.unlock();
        }
    }

    public void cancel(String ruleId) {
        try {
            lock.lock();
            if (StringUtils.isNotBlank(ruleId)) {
                ScheduledFuture<?> future = taskMap.get(ruleId);
                if (future != null) {
                    future.cancel(true);
                    taskMap.remove(ruleId);
                    alertStatMap.remove(ruleId);
                    log.info("取消了alertJob任务:{}", ruleId);
                }
            }
        } catch (Exception ex) {
            log.error(ex.getMessage(), ex);
        } finally {
            lock.unlock();
        }
    }

    /**
     * 创建job
     *
     * @param alertRuleVo
     *
     * @return
     */
    private Runnable buildAlertJob(AlertRuleVo alertRuleVo) {
        return new Runnable() {
            @Override
            public void run() {
                log.info("报警任务{}-{}于{}触发执行", alertRuleVo.getRuleId(), alertRuleVo.getDescribe(), DateTools.getSysDate19());
                Date today = DateTools.getSysDate();
                long toTime = today.getTime();
                //统计周期内
                long formTime = toTime - TimeUnit.SECONDS.toMillis(alertRuleVo.getStatSecondsAgo());
                BoolQueryBuilder boolQueryBuilder = getBoolQueryBuilder(toTime, formTime);
                int actualErrorCount = (int)ESFactory.getInstance().count(EAGLE_MONITOR_ES_INDEX, EAGLE_MONITOR_ES_TYPE, boolQueryBuilder);
                //记录连续触发报警的次数
                AlertStat alertStat = handleStatResult(actualErrorCount);
                log.info("报警任务{}-返回结果:{}条，连续异常次数:{}次", alertRuleVo.getRuleId(), actualErrorCount,alertStat.getCurrentProblemPeriod());
                //判断是否满足触发报警阈值
                if (actualErrorCount >= alertRuleVo.getTriggerCount()) {
                    log.info(String.format("符合报警规则[%s]:%s",alertRuleVo.getRuleId(),alertRuleVo.toString()));
                    List<RpcTraceInfoVo> errorList = getRpcTraceInfoVoList(boolQueryBuilder);
                    if (errorList != null && errorList.size() > 0) {
                        String alertLogId = UUID.randomUUID().toString().toLowerCase().replaceAll("-", "");
                        //TODO:应重构抽象成Notification
                        //写报警日志
                        saveAlertLog(alertLogId, formTime, toTime, actualErrorCount, errorList, alertStat);
                        //邮件报警
                        sendEmails(alertLogId, formTime, toTime, actualErrorCount, errorList, alertStat);
                        //企业微信报警
                        sendWxWorkMsg(alertLogId, formTime, toTime, actualErrorCount, errorList, alertStat);
                    }
                }else{
                    RecoverMsgLog recoverMsgLog = alertStat.takeRecoverMsgLog();
                    //判断是否需要故障恢复发送通知
                    if (recoverMsgLog != null) {
                        try {
                            //卡片版
                            String subject = "EagleEye-Alert:";
                            if (StringUtils.isNotBlank(alertRuleVo.getLevel())) {
                                subject += String.format("[%s]-",alertRuleVo.getLevel());
                            }
                            subject += String.format("[%s]恢复正常通知", alertRuleVo.getDescribe());
                            String body = String.format("当前已连续%s次监控正常，该故障已于%s恢复正常。(之前已连续告警%s次，总计涉及%s条。)",
                                    recoverMsgLog.getHealthCount(),
                                    DateTools.FormatDate19(new Date(formTime)),
                                    recoverMsgLog.getProblemCount(),
                                    recoverMsgLog.getActualErrorSum()
                            );
                            String emails = extractAllReceivers(alertRuleVo.getEmails());
                            //发送邮件
                            if(StringUtils.isNotBlank(emails)) {
                                String emailBody = "["+alertRuleVo.getDescribe()+"]-"+body ;
                                sendMailUtils.sendMail(subject, "EagleEye-Alert", emails, emailBody, null);
                            }
                            String qywxEmails = extractAllReceivers(alertRuleVo.getQywxEmails());
                            //发送企业微信
                            if(StringUtils.isNotBlank(qywxEmails)) {
                                String qywxBody = "<div class='highlight'>" + body + "</div>";
                                wxWorkUtils.sendCard(qywxEmails, subject, "", qywxBody);
                            }
                        } catch (Exception ex) {
                            log.error(ex.getMessage(),ex);
                        }
                    }
                }
            }
            /**
             * 处理统计结果
             */
            private AlertStat handleStatResult(int actualErrorCount) {
                AlertStat alertStat = alertStatMap.get(alertRuleVo.getRuleId());
                if(alertStat==null){
                    alertStat = new AlertStat(alertRuleVo.getRuleId(),alertRuleVo.getProblemMinCount(),alertRuleVo.getHealthMinCount());
                    alertStatMap.put(alertRuleVo.getRuleId(), alertStat);
                }

                //判断是否满足触发报警阈值
                if (actualErrorCount >= alertRuleVo.getTriggerCount()) {
                    alertStat.incrementProblemPeriod(actualErrorCount);
                }else{
                    alertStat.incrementHealthPeriod();
                }

                return alertStat;
            }
            /**
             * 提取需要发送的email
             * @param emails 当前发送的邮箱列表，格式：邮箱#最低连续故障次数,邮箱#最低连续故障次数, ...
             */
            private String extractAllReceivers(String emails) {
                return extractReceivers(emails,null);
            }

            /**
             * 提取需要发送的email
             * @param emails 当前发送的邮箱列表，格式：邮箱#最低连续故障次数,邮箱#最低连续故障次数, ...
             * @param alertStat 当前发生的连续故障周期次数
             */
            private String extractReceivers(String emails,AlertStat alertStat) {
                StringBuilder receives = new StringBuilder();
                if(StringUtils.isNotBlank(emails)) {
                    String[] emailAndProblemMinCountArr = StringUtils.split(emails, ",");
                    for(String emailAndProblemMinCount : emailAndProblemMinCountArr) {
                        String[] split = StringUtils.split(emailAndProblemMinCount, "#");
                        if(receives.length()>0) {
                            receives.append(",");
                        }
                        if (split.length == 2) {
                            String email = split[0];
                            int minCount = Integer.parseInt(split[1]);
                            if(!StringUtils.endsWithIgnoreCase(email,"@credithc.com")){
                                email +=  "@credithc.com";
                            }
                            if(alertStat!=null){
                                if (alertStat.getCurrentProblemPeriod() >= minCount || alertStat.isProblemStatus()) {
                                    receives.append(email);
                                }
                            }else{
                                receives.append(email);
                            }
                        }else{
                            receives.append(emailAndProblemMinCount);
                        }

                    }
                }
                return receives.toString();
            }
            /**
             * 发送企业微信报警
             */
            private void sendWxWorkMsg(String alertLogId, long formTime, long toTime, int actualErrorCount, List<RpcTraceInfoVo> errorList, AlertStat alertStat) {
                String receivers = extractReceivers(alertRuleVo.getQywxEmails(), alertStat);
                if(StringUtils.isNotBlank(receivers)) {
                    try {
                        //文字明细版
                        /*
                        StringBuffer sb = new StringBuffer();
                        String subject = String.format("EagleEye-Alert:%s[%s]-[%s]报警信息\n", alertRuleVo.getPlatformName(), alertRuleVo.getEnv(), alertRuleVo.getDescribe());
                        sb.append(subject);
                        //前5条明细
                        int minSize = Math.min(5, list.size());
                        for(int i = 0; i < minSize; i++) {
                            RpcTraceInfoVo traceInfoVo = list.get(i);
                            sb.append("第" + (i + 1) + "条:\n");
                            sb.append("调用链ID:" + traceInfoVo.getTraceId() + "\n");
                            sb.append("客户端:" + traceInfoVo.getClientName() + "\n");
                            sb.append("服务名:" + traceInfoVo.getServiceName() + "\n");
                            sb.append("方法名:" + traceInfoVo.getMethodName() + "\n");
                            sb.append("结果:" + traceInfoVo.getResult() + "\n");
                            sb.append("耗时:" + traceInfoVo.getRunTime() + "ms\n");
                            sb.append("返回值:" + StringUtils.left(traceInfoVo.getResponseJson(), 100) + "...\n");
                        }
                        String errorJson = sb.toString();
                        String receivers = alertRuleVo.getQywxEmails();
                        wxWorkUtils.sendText(receivers, errorJson);
                        */
                        //卡片版
                        String subject = "EagleEye-Alert:";
                        if (StringUtils.isNotBlank(alertRuleVo.getLevel())) {
                            subject += String.format("[%s]-",alertRuleVo.getLevel());
                        }
                        subject += String.format("[%s]", alertRuleVo.getDescribe());

                        int currentProblemPeriod = alertStat.getCurrentProblemPeriod();
                        if(currentProblemPeriod >1){
                            subject += "-连续"+currentProblemPeriod+"次";
                        }
                        String body = String.format("<div class='gray'>统计时间范围:%s至%s</div><div class='highlight'>触发了[%s]规则,涉及总计条数：%s条。</div>",
                                DateTools.FormatDate19(new Date(formTime)),
                                DateTools.FormatDate19(new Date(toTime)),
                                alertRuleVo.getDescribe(),
                                actualErrorCount);
                        //判断是否需要发送故障恢复通知
                        if(alertStat.isProblemStatus()){
                            body+=String.format("<div>(当前已连续告警%s次，故障恢复后会给您发送恢复通知)</div>",currentProblemPeriod);
                        }
                        boolean retCard = wxWorkUtils.sendCard(receivers, subject, alertLogId,body);
                    } catch (Exception ex) {
                        log.error(ex.getMessage(),ex);
                    }
                }
            }
            /**
             * 发送邮件报警
             */
            private void sendEmails(String alertLogId,long formTime, long toTime, int actualErrorCount, List<RpcTraceInfoVo> errorList, AlertStat alertStat) {
                String receivers = extractReceivers(alertRuleVo.getEmails(), alertStat);
                if (StringUtils.isNotBlank(receivers)) {
                    try {
                        StringBuffer sb = new StringBuffer();
                        String format_stat = String.format("统计时间范围:%s至%s,触发了[%s]规则,涉及总计条数：" + actualErrorCount + "条。\n",
                                DateTools.FormatDate19(new Date(formTime)),
                                DateTools.FormatDate19(new Date(toTime)),
                                alertRuleVo.getDescribe());
                        sb.append(format_stat);
                        //判断是否需要发送故障恢复通知
                        int currentProblemPeriod = alertStat.getCurrentProblemPeriod();
                        if (alertStat.isProblemStatus()) {
                            sb.append(String.format("(当前已连续告警%s次，故障恢复后会给您发送恢复通知)\n", currentProblemPeriod));
                        }
                        //发送邮件
                        for(int i = 0; i < errorList.size(); i++) {
                            sb.append("第" + (i + 1) + "条:" + FastJsonConvert.convertObjectToJsonWithPrettyFormat(errorList.get(i)) + "\n");
                        }
                        String errorJson = sb.toString();
                        String subject = "EagleEye-Alert:";
                        if (StringUtils.isNotBlank(alertRuleVo.getLevel())) {
                            subject += String.format("[%s]-",alertRuleVo.getLevel());
                        }
                        subject += String.format("[%s]-[%s]",  alertRuleVo.getEnv(), alertRuleVo.getDescribe());

                        if (currentProblemPeriod > 1) {
                            subject += "-连续" + currentProblemPeriod + "次";
                        }
                        sendMailUtils.sendMail(subject, "EagleEye-Alert", receivers, errorJson, null);
                    } catch (Exception ex) {
                        log.error(ex.getMessage(), ex);
                    }
                }
            }
            /**
             * 保存报警日志
             */
            private void saveAlertLog(String alertLogId, long formTime, long toTime, int actualErrorCount, List<RpcTraceInfoVo> errorList,AlertStat alertStat) {
                try{
                    AlertLogVo alertLogVo = new AlertLogVo();
                    alertLogVo.setAlertId(alertLogId);
                    alertLogVo.setRuleId(alertRuleVo.getRuleId());
                    alertLogVo.setAlertTime(ISO_DATETIME_TIME_ZONE_FORMAT_WITH_MILLIS.format(new Date()));
                    alertLogVo.setEmails(alertRuleVo.getEmails());
                    alertLogVo.setQywxEmails(alertRuleVo.getQywxEmails());
                    alertLogVo.setEnv(alertRuleVo.getEnv());
                    alertLogVo.setTriggerCount(alertRuleVo.getTriggerCount());
                    alertLogVo.setActualCount(actualErrorCount);
                    alertLogVo.setPlatformName(alertRuleVo.getPlatformName());
                    alertLogVo.setQueryString(alertRuleVo.getQueryString());
                    alertLogVo.setProblemPeriod(alertStat.getCurrentProblemPeriod());
                    alertLogVo.setLevel(alertRuleVo.getLevel());
                    String title = String.format("统计时间范围:%s至%s,触发了[%s]规则,涉及总计条数：" + actualErrorCount + "条。\n",
                            DateTools.FormatDate19(new Date(formTime)),
                            DateTools.FormatDate19(new Date(toTime)),
                            alertRuleVo.getDescribe());
                    alertLogVo.setTitle(title);
                    StringBuffer messageBuffer = new StringBuffer();
                    //显示前10条
                    int min = Math.min(10, errorList.size());
                    for(int i = 0; i < min; i++) {
                        RpcTraceInfoVo traceInfoVo = errorList.get(i);
                        messageBuffer.append("第" + (i + 1) + "条:\n");
                        messageBuffer.append("调用链ID:" + traceInfoVo.getTraceId() + "\n");
                        messageBuffer.append("客户端:" + traceInfoVo.getClientName() + "\n");
                        messageBuffer.append("服务名:" + traceInfoVo.getServiceName() + "\n");
                        messageBuffer.append("方法名:" + traceInfoVo.getMethodName() + "\n");
                        messageBuffer.append("结果:" + traceInfoVo.getResult() + "\n");
                        messageBuffer.append("耗时:" + traceInfoVo.getRunTime() + "ms\n");
                        messageBuffer.append("返回值:" + StringUtils.left(traceInfoVo.getResponseJson(), 100) + (StringUtils.length(traceInfoVo.getResponseJson())>100?"...":""));
                        messageBuffer.append("\n\n");
                    }
                    alertLogVo.setMessage(messageBuffer.toString());
                    ESFactory.getInstance().insertDocument(EAGLE_ALERT_LOG_ES_INDEX, EAGLE_ALERT_LOG_ES_TYPE, alertLogVo);
                }catch(Exception ex){
                    log.error(ex.getMessage(),ex);
                }
            }

            /**
             * 获取查询条件
             */
            private BoolQueryBuilder getBoolQueryBuilder(long toTime, long formTime) {
                //搜索条件
                BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
                boolQueryBuilder.must(QueryBuilders.matchPhraseQuery("platformName.raw", alertRuleVo.getPlatformName()));
                if (formTime != 0) {
                    boolQueryBuilder.must(QueryBuilders.rangeQuery("requestDateTime").from(formTime));
                }
                if (toTime != 0) {
                    boolQueryBuilder.must(QueryBuilders.rangeQuery("requestDateTime").to(toTime));
                }
                boolQueryBuilder.must(QueryBuilders.matchPhraseQuery("env", alertRuleVo.getEnv()));
                if (StringUtils.isNotBlank(alertRuleVo.getQueryString())) {
                    QueryStringQueryBuilder queryStringQueryBuilder = QueryBuilders.queryStringQuery(alertRuleVo.getQueryString());
                    boolQueryBuilder.must(queryStringQueryBuilder);
                } else {
                    boolQueryBuilder.must(QueryBuilders.matchPhraseQuery("result", "ERROR"));
                }
                //排除定时任务
                boolQueryBuilder.mustNot(QueryBuilders.matchPhraseQuery("serviceName", "taskIceService"));
                return boolQueryBuilder;
            }

            /**
             * 获取错误明细
             */
            private List<RpcTraceInfoVo> getRpcTraceInfoVoList(BoolQueryBuilder boolQueryBuilder) {
                List<SortBuilder> sortBuilderList = new ArrayList<>();
                SortBuilder sortBuilder0 = SortBuilders.fieldSort("requestDateTime").order(SortOrder.DESC);
                sortBuilderList.add(sortBuilder0);
                int pageSize = 10;
                int from = 0;
                List<RpcTraceInfoVo> list =
                        ESFactory.getInstance().search(
                                RpcTraceInfoVo.class,
                                EAGLE_MONITOR_ES_INDEX,
                                EAGLE_MONITOR_ES_TYPE,
                                boolQueryBuilder,
                                sortBuilderList,
                                from,
                                pageSize);
                for(RpcTraceInfoVo vo : list) {
                    vo.setRequestDateTime(vo.getRequestDateTime().replace("T", " ").substring(0, 23));
                }
                return list;
            }
        };
    }

}
class AlertStat {
    private String ruleId;
    /**到达故障状态的连续次数*/
    private int problemMinCount;
    /**从故障状态转为健康状态的连续次数*/
    private int healthMinCount;

    public AlertStat(String ruleId, int problemMinCount, int healthMinCount) {
        this.ruleId = ruleId;
        this.problemMinCount = problemMinCount;
        this.healthMinCount = healthMinCount;
    }

    /**连续报警周期的次数*/
    private AtomicInteger problemPeriod =new AtomicInteger(0);
    /**连续报警周期时的匹配结果总数*/
    private AtomicInteger actualErrorSum =new AtomicInteger(0);
    /**连续正常周期次数*/
    private AtomicInteger healthPeriod =new AtomicInteger(0);
    /**需要发送故障恢复的通知*/
    private RecoverMsgLog recoverMsgLog =null;

    public boolean isProblemStatus() {
        return problemMinCount > 0 && (problemPeriod.get() >= problemMinCount);
    }

    //增加一次失败周期，正常周期清零
    void incrementProblemPeriod(int actualErrorCount) {
        this.problemPeriod.incrementAndGet();
        this.healthPeriod.set(0);
        this.actualErrorSum.addAndGet(actualErrorCount);
    }

    //增加一次正常周期 ，如果当前处于故障状态，需要判断是否达到了连续健康次数
    void incrementHealthPeriod() {
        int healthCount = this.healthPeriod.incrementAndGet();
        //判断当前是否已经到达了problem状态
        if(isProblemStatus()){
            //判断当前是否可以转为健康状态
            if(healthCount>= healthMinCount) {
                recoverMsgLog = new RecoverMsgLog(healthCount, problemPeriod.get(), actualErrorSum.get());
                this.problemPeriod.set(0);
                this.actualErrorSum.set(0);
            }
        }else{
            this.problemPeriod.set(0);
            this.actualErrorSum.set(0);
        }
    }

    int getCurrentProblemPeriod() {
        return this.problemPeriod.get();
    }

    /**
     * 获取并清空需要发送的故障恢复通知
     */
    RecoverMsgLog takeRecoverMsgLog() {
        RecoverMsgLog lastRecoverMsg = recoverMsgLog;
        recoverMsgLog = null;
        return lastRecoverMsg;
    }
}
@Data
class RecoverMsgLog {
    private int problemCount;
    private int healthCount;
    private int actualErrorSum;

    public RecoverMsgLog(int healthCount, int problemCount, int actualErrorSum) {
        this.healthCount = healthCount;
        this.problemCount = problemCount;
        this.actualErrorSum = actualErrorSum;
    }
}
@Data
class Receiver{
    /**邮箱*/
    private String email;
    /**连续触发多少次*/
    private int problemMinCount;
}
